package org.example.common.handle;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.example.service.mqtt.TopicService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @Author 李荣
 * @Date 2025/1/3
 * @Description
 */

@Component
@Slf4j
public class MqttServiceHandle implements MqttCallback {

    private static final int MAX_RECONNECT_ATTEMPTS = 5;
    private static final long RECONNECT_DELAY_S = 5;

    private final MqttClient mqttClient;

    private final TopicService topicService;

//    @Autowired
//    private KafkaTemplate<String,Object> kafkaTemplate;
    @Autowired
    private WebSocketHandle webSocketHandle;

    @Autowired
    public MqttServiceHandle(MqttClient mqttClient, TopicService topicService)
    {
        this.mqttClient  = mqttClient;
        this.topicService = topicService;
    }

    public void publish(String message,String topic) throws MqttException {
        MqttMessage mqttMessage = new MqttMessage(message.getBytes());
        mqttMessage.setQos(2);
        mqttClient.publish(topic,mqttMessage);
    }

    //连接断开
    @Override
    public void connectionLost(Throwable throwable) {
        log.info("连接断开，正在重试.... {}",throwable.getMessage());
        reconnect();
    }



    //接收消息
    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {

        log.info("s: {}",topic);
        byte[] payload = mqttMessage.getPayload();

        log.info("message: {}",new String(payload));
        webSocketHandle.sendToTopic(new String(payload));
    }

    //消息发送完成
    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        log.info("消息的传递完成");
    }
    private void reconnect() {

        int retryCount = 0;

        while (!mqttClient.isConnected())
        {
            try {
                mqttClient.connect();
                List<String> topicsToSubscribe = topicService.getTopicsToSubscribe();
                for (String topic : topicsToSubscribe) {
                    mqttClient.subscribe(topic);
                }
            }catch (Exception e)
            {
                retryCount ++;
                log.info("正在重试，重试第{}次",retryCount);
                //休眠之后在重试
                try {
                    wait(5000);
                } catch (InterruptedException ex) {
                    e.printStackTrace();
                }
            }
        }

    }

}
